/*
    服务器模块：利用muduo、以及业务层模块（虚拟机模块、消费者管理模块、信道管理模块）进行整合，搭建一个服务器
*/
#pragma once
#include "proto/codec.h"
#include "proto/dispatcher.h"

#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "virtual_host.hpp"
#include "consumer.hpp"
#include "connection.hpp"

namespace mq
{
    #define DBFILE "meta.db" // sqlite3 数据库文件
    #define VIRTUAL_HOST_NAME "default_virtual_host"

    // 服务器
    class BrokerServer
    {
    public:
        BrokerServer(const std::string &data_dir, int port, const std::string &ip = "0.0.0.0") :
            _server(&_base_loop, muduo::net::InetAddress(ip, port), "server", muduo::net::TcpServer::kReusePort),
            _dispatcher(std::bind(&BrokerServer::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
            _codec_ptr(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
            _virtual_host_ptr(std::make_shared<VirtualHost>(VIRTUAL_HOST_NAME, data_dir + "/" + DBFILE, data_dir)),
            _consumer_manager_ptr(std::make_shared<ConsumerManager>()),
            _connection_manager_ptr(std::make_shared<ConnectionManager>()),
            _thread_pool_ptr(std::make_shared<ThreadPool>())
        {
            // 1. 给消息分发器注册业务处理函数
            _dispatcher.registerMessageCallback<OpenChannelRequest>(std::bind(&BrokerServer::onOpenChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<CloseChannelRequest>(std::bind(&BrokerServer::onCloseChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<DeclareExchangerRequest>(std::bind(&BrokerServer::onDeclareExchanger, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<RemoveExchangerRequest>(std::bind(&BrokerServer::onRemoveExchanger, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<DeclareQueueRequest>(std::bind(&BrokerServer::onDeclareMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<RemoveQueueRequest>(std::bind(&BrokerServer::onRemoveMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<BindRequest>(std::bind(&BrokerServer::onBindMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<UnbindRequest>(std::bind(&BrokerServer::onUnbindMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<PublishRequest>(std::bind(&BrokerServer::onPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<AckRequest>(std::bind(&BrokerServer::onAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<ConsumeRequest>(std::bind(&BrokerServer::onConsumer, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<CancelRequest>(std::bind(&BrokerServer::onCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            // 2. 给muduo服务器设置连接建立和断开时的回调 & 消息回调
            _server.setConnectionCallback(std::bind(&BrokerServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec_ptr.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void start()
        {
            INF_LOG("BrokerServer start...\n");
            _server.start();
            _base_loop.loop();
        }

    private:
        /*
            下面都是业务回调函数，用于注册给Dispatcher消息分发器，当接收到不同的消息类型的时候，选择对应的回掉函数来处理
        */
        // 打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const OpenChannelRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while openning channel, connection not found\n");
                conn->shutdown(); // 非法连接，直接关闭
                return;
            }

            // 2. 打开信道（创建信道）并响应
            DBG_LOG("open channel success, channel id: %s, request id: %s\n", req->channel_id().c_str(), req->request_id().c_str());
            connection_ptr->openChannel(req);
        }

        // 关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const CloseChannelRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while closing channel, connection not found\n");
                conn->shutdown(); // 非法连接，直接关闭
                return;
            }

            // 2. 关闭信道（销毁信道）并响应
            DBG_LOG("close channel success, channel id: %s, request id: %s\n", req->channel_id().c_str(), req->request_id().c_str());
            connection_ptr->closeChannel(req);
        }

        // 声明交换机
        void onDeclareExchanger(const muduo::net::TcpConnectionPtr& conn, const DeclareExchangerRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while declaring exchanger, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while declaring exchanger, channel not found\n");
                return;
            }

            // 3. 声明交换机并响应
            DBG_LOG("declare exchanger success, name: %s, type: %d, durable: %d, request id: %s\n", req->exchanger_name().c_str(), req->exchanger_type(), req->durable(), req->request_id().c_str());
            channel_ptr->declareExchanger(req);
        }

        // 删除交换机
        void onRemoveExchanger(const muduo::net::TcpConnectionPtr& conn, const RemoveExchangerRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while removing exchanger, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while removing exchanger, channel not found\n");
                return;
            }

            // 3. 删除交换机并响应
            channel_ptr->removeExchanger(req);
        }

        // 声明消息队列
        void onDeclareMsgQueue(const muduo::net::TcpConnectionPtr& conn, const DeclareQueueRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while declaring message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while declaring message queue, channel not found\n");
                return;
            }

            // 3. 声明消息队列并响应
            DBG_LOG("declare message queue success, message queue name: %s, request id: %s\n", req->queue_name().c_str(), req->request_id().c_str());
            channel_ptr->declareMsgQueue(req);
        }

        // 删除消息队列
        void onRemoveMsgQueue(const muduo::net::TcpConnectionPtr& conn, const RemoveQueueRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while removing message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while removing message queue, channel not found\n");
                return;
            }

            // 3. 删除消息队列并响应
            channel_ptr->removeMsgQueue(req);
        }

        // 绑定消息队列
        void onBindMsgQueue(const muduo::net::TcpConnectionPtr& conn, const BindRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while binding message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while binding message queue, channel not found\n");
                return;
            }

            // 3. 绑定消息队列并响应
            DBG_LOG("bind message queue success, exchanger name: %s, message queue name: %s, request id: %s\n", 
                req->exchanger_name().c_str(), req->queue_name().c_str(), req->request_id().c_str());
            channel_ptr->bind(req);
        }

        // 解除绑定消息队列
        void onUnbindMsgQueue(const muduo::net::TcpConnectionPtr& conn, const UnbindRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while unbinding message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while buninding message queue, channel not found\n");
                return;
            }

            // 3. 解除绑定消息队列并响应
            channel_ptr->unbind(req);
        }

        // 发布消息
        void onPublish(const muduo::net::TcpConnectionPtr& conn, const PublishRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while publishing message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while publishing message queue, channel not found\n");
                return;
            }

            // 3. 发布消息并响应
            channel_ptr->publish(req);
        }

        // 应答消息
        void onAck(const muduo::net::TcpConnectionPtr& conn, const AckRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while acking message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while acking message queue, channel not found\n");
                return;
            }

            // 3. 应答消息并响应
            channel_ptr->ack(req);
        }

        // 订阅消息队列
        void onConsumer(const muduo::net::TcpConnectionPtr& conn, const ConsumeRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while consuming message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while consuming message queue, channel not found\n");
                return;
            }

            // 3. 应答消息并响应
            channel_ptr->consume(req);
        }

        // 取消订阅消息队列
        void onCancel(const muduo::net::TcpConnectionPtr& conn, const CancelRequestPtr& req, muduo::Timestamp)
        {
            // 1. 查找连接是否存在
            Connection::ConnectionPtr connection_ptr = _connection_manager_ptr->getConnection(conn);
            if(connection_ptr == nullptr)
            {
                DBG_LOG("while canceling message queue, connection not found\n");
                conn->shutdown();
                return;
            }

            // 2. 查找连接上的信道是否存在
            Channel::ChannelPtr channel_ptr = connection_ptr->getChannel(req->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("while canceling message queue, channel not found\n");
                return;
            }

            // 3. 取消订阅并响应
            channel_ptr->cancel(req);
        }

        // 当出现未知业务请求时，默认处理函数
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const ::MessagePtr& message, muduo::Timestamp)
        {
            DBG_LOG("unknown message: %s", message->GetTypeName().c_str());
            conn->shutdown();
        }

        // 当连接建立和断开时的回调
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                // 新连接建立，创建一个自定义连接，并维护
                _connection_manager_ptr->newConnection(_consumer_manager_ptr, conn, _codec_ptr, _virtual_host_ptr, _thread_pool_ptr);
            }
            else
            {
                // 连接断开，删除自定义连接（自定义连接删除，其管理的所有信道也会删除，信道删除，通过该信道的订阅者也会自动删除）
                _connection_manager_ptr->delConnection(conn);
            }
        }

    private:
        muduo::net::EventLoop _base_loop;                                   // 时间循环
        muduo::net::TcpServer _server;                                      // 服务器
        ProtobufDispatcher _dispatcher;                                     // 消息分发器
        ProtobufCodecPtr _codec_ptr;                                        // 协议处理器
        VirtualHost::VirtualHostPtr _virtual_host_ptr;                      // 虚拟主机句柄
        ConsumerManager::ConsumerManagerPtr _consumer_manager_ptr;          // 消费者管理句柄
        ConnectionManager::ConnectionManagerPtr _connection_manager_ptr;    // 连接管理句柄
        ThreadPool::ThreadPoolPtr _thread_pool_ptr;                         // 线程池句柄
    };
}